package com.google.appengine.tools.mapreduce.inputs;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

import com.google.appengine.api.blobstore.BlobKey;
import com.google.appengine.api.blobstore.BlobstoreInputStream;
import com.google.appengine.tools.mapreduce.Input;
import com.google.appengine.tools.mapreduce.InputReader;
import com.google.appengine.tools.mapreduce.KeyValue;
import com.google.common.base.Preconditions;
import com.google.common.io.CountingInputStream;

/**
 * BlobstoreZipLineInput shards zip files in Blobstore on zip entries.
 *
 */
public class BlobstoreZipInput extends Input<KeyValue<String, byte[]>> {

	private static final long serialVersionUID = -322743073479038974L;

	private static final Logger log = Logger.getLogger(BlobstoreZipInput.class.getName());

	private final String blobKey;
	private final byte separator;
	private final int shardCount;

	public BlobstoreZipInput(String blobKey, byte separator, int shardCount) {
		this.blobKey = blobKey;
		this.separator = separator;
		this.shardCount = shardCount;
	}

	@Override
	public List<? extends InputReader<KeyValue<String, byte[]>>> createReaders()
			throws IOException {
	    Preconditions.checkNotNull(blobKey);
	    Preconditions.checkArgument(shardCount > 0);

	    // TODO Use shardCount to determine how many readers to create.

		CountingInputStream countingInput = new CountingInputStream(
				new BlobstoreInputStream(new BlobKey(blobKey)));
		ZipInputStream zipInput = new ZipInputStream(countingInput);
		try {
			List<BlobstoreZipInputReader> readers =
					new ArrayList<BlobstoreZipInputReader>();
			ZipEntry zipEntry;
			byte[] buffer = new byte[1024];
			long offset = 0, endOffset = 0;
			while ((zipEntry = zipInput.getNextEntry()) != null) {
				// TODO(ldee): Determine better way to get end offset
				while (zipInput.read(buffer) > 0) {
					// Do nothing, but this step is needed
					// to make zipEntry.getCompressedSize() work.
					// Otherwise, it returns -1.
				}
				endOffset = offset
						+ 30L // header
						+ zipEntry.getName().getBytes("utf-8").length
						+ zipEntry.getCompressedSize()
						+ (zipEntry.getExtra() == null ? 0L : zipEntry.getExtra().length)
						+ 16L; // end
				if (!zipEntry.isDirectory()) {
					log.info(String.format(
							"Added BlobstoreZipInputReader for %s", zipEntry.getName()));
					readers.add(new BlobstoreZipInputReader(
							blobKey, offset, endOffset, separator,
							zipEntry));
				}
				offset = endOffset;
			}
			return readers;
		} finally {
			zipInput.closeEntry();
			zipInput.close();
		}
	}

}
